package org.ala.distributed_transaction_commons.redis;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import org.ala.distributed_transaction_commons.AbstractDistributedTransaction;
import org.ala.distributed_transaction_commons.DistributedTransactionRole;
import org.ala.distributed_transaction_commons.DistributedTransactionStatus;
import org.ala.distributed_transaction_commons.IDistributedTransaction;
import org.ala.distributed_transaction_commons.configure.DistributedTransactionConfigure;
import org.ala.distributed_transaction_commons.redis.datas.TransactionContext;
import org.ala.distributed_transaction_commons.redis.datas.TransactionParticipant;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.StringUtils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Pipeline;


/**
 * redis工具
 *
 * @author 骆毅(ala)
 * @date 2024年2月17日
 */
public class DistributedTransactionRedisTools implements InitializingBean {

	
	/**
	 * 相关配置
	 */
	protected DistributedTransactionConfigure.RedisConfigure conf;
	
	/**
	 * redis pool
	 */
	protected JedisPool pool;
	/**
	 * 日期格式化工具
	 */
	protected DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
	
	
	@Override
	public void afterPropertiesSet() throws Exception {
		//	初始化jedispool
		initJedisPool();
	}
	/**
	 * 初始化JedisPool
	 */
	protected void initJedisPool() {
		GenericObjectPoolConfig<Jedis> poolConfig = new GenericObjectPoolConfig<Jedis>();
		poolConfig.setMinIdle(conf.getPoolIdle());
		poolConfig.setMaxTotal(conf.getPoolMax());
		pool = new JedisPool(poolConfig, conf.getHost(), conf.getPort(), conf.getTimeout(), conf.getPassword(), conf.getDatabase());
	}
	
	
	/**
	 * 在jedis作用于内工作
	 */
	public <T> T withJedis(ICallable<T> callable) {
		Jedis jedis = pool.getResource();
		try {
			return callable.call(jedis);
		} finally {
			pool.returnResource(jedis);
		}
	}
	public static interface ICallable<T> {
		T call(Jedis jedis);
	}
	
	
	
	/****************************************************************************************************
	 * 
	 * 事务环境相关操作
	 * 
	 ****************************************************************************************************/
	/**
	 * 初始化事务环境
	 * @param firstTx	第一个被启动的事务（第一个参与者）
	 */
	public void initTransactionContext(AbstractDistributedTransaction firstTx) {
		withJedis(jedis -> {
			Pipeline tx = jedis.pipelined();
			String txRedisId = KeyNamespace.transactionContextId(firstTx.biz(), firstTx.id());
			//	初始化context
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_ID, firstTx.id());
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_BIZ, firstTx.biz());
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_STATUS, Integer.toString(firstTx.status().getVal()));
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_MIN_PARTICIPANT_COUNT_BEGIN, Integer.toString(firstTx.getMinParticipantCountBegin()));
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_MIN_PARTICIPANT_COUNT_CAN_COMMIT, Integer.toString(firstTx.getMinParticipantCountCanCommit()));
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_MIN_PARTICIPANT_COUNT_PRE_COMMIT, Integer.toString(firstTx.getMinParticipantCountPreCommit()));
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_MIN_PARTICIPANT_COUNT_COMMIT, Integer.toString(firstTx.getMinParticipantCountCommit()));
			long now = firstTx.startTimestamp();
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_START_TIMESTAMP, Long.toString(now));
			tx.hset(txRedisId, KeyNamespace.FIELD_CONTEXT_START_DATE, dateFormat.format(new Date(now)));
			tx.expire(txRedisId, KeyNamespace.TIMEOUT);
			
			//	保存参与者
			saveParticipant(tx, firstTx);
			tx.sync();
			return null;
		});
	}
	/**
	 * 查看事务环境当前状态
	 */
	public DistributedTransactionStatus getContextStatus(String biz, String txId) {
		String val = withJedis(jedis -> {
			return jedis.hget(KeyNamespace.transactionContextId(biz, txId), KeyNamespace.FIELD_CONTEXT_STATUS);
		});
		if (StringUtils.isEmpty(val)) {return null;}
		return DistributedTransactionStatus.getInstance(Integer.parseInt(val));
	}
	/**
	 * 修改事务环境当前状态
	 */
	public void updateContextStatus(Pipeline tx, String biz, String txId, DistributedTransactionStatus status) {
		tx.hset(KeyNamespace.transactionContextId(biz, txId), KeyNamespace.FIELD_CONTEXT_STATUS, Integer.toString(status.getVal()));
	}
	/**
	 * 修改事务环境当前状态
	 */
	public void updateContextStatus(String biz, String txId, DistributedTransactionStatus status) {
		withJedis(jedis -> jedis.hset(KeyNamespace.transactionContextId(biz, txId), KeyNamespace.FIELD_CONTEXT_STATUS, Integer.toString(status.getVal())));
	}
	/**
	 * 判断事务环境是否存在
	 */
	public boolean existsContext(String biz, String txId) {
		return withJedis(jedis -> {
			return jedis.exists(KeyNamespace.transactionContextId(biz, txId));
		});
	}
	/**
	 * 取当前事务环境
	 */
	public TransactionContext getTransactionContext(String biz, String txId) {
		//	取context
		Map<String, String> map = withJedis(jedis -> {
			//	取线程环境
			String redisContextId = KeyNamespace.transactionContextId(biz, txId);
			return jedis.hgetAll(redisContextId);
		});
		if (map == null || map.isEmpty()) {return null;}
		
		TransactionContext context = new TransactionContext();
		context.setId(txId);
		context.setBiz(map.get(KeyNamespace.FIELD_CONTEXT_BIZ));
		context.setStatus(DistributedTransactionStatus.getInstance(Integer.parseInt(map.get(KeyNamespace.FIELD_CONTEXT_STATUS))));
		context.setStartTimestamp(Long.parseLong(map.get(KeyNamespace.FIELD_CONTEXT_START_TIMESTAMP)));
		context.setMinParticipantCountBegin(Integer.parseInt(map.get(KeyNamespace.FIELD_CONTEXT_MIN_PARTICIPANT_COUNT_BEGIN)));
		context.setMinParticipantCountCanCommit(Integer.parseInt(map.get(KeyNamespace.FIELD_CONTEXT_MIN_PARTICIPANT_COUNT_CAN_COMMIT)));
		context.setMinParticipantCountPreCommit(Integer.parseInt(map.get(KeyNamespace.FIELD_CONTEXT_MIN_PARTICIPANT_COUNT_PRE_COMMIT)));
		context.setMinParticipantCountCommit(Integer.parseInt(map.get(KeyNamespace.FIELD_CONTEXT_MIN_PARTICIPANT_COUNT_COMMIT)));
		//	取参与者id集合
		context.setParticipantIds(withJedis(jedis -> {
			return jedis.smembers(KeyNamespace.transactionParticipantSetIds(biz, txId));
		}));
		//	取参与者集合
		if (context.getParticipantIds() != null && !context.getParticipantIds().isEmpty()) {
			context.setParticipantSet(getParticipant(biz, context.getParticipantIds().toArray(new String[0])));
		}
		
		return context;
	}
	/**
	 * 清空事务环境
	 */
	public void clearTransactionContext(String biz, String txId) {
		//	事务环境redisId，参与者集合id
		String contextRedisId = KeyNamespace.transactionContextId(biz, txId);
		String participantSetId = KeyNamespace.transactionParticipantSetIds(biz, txId);
		//	取参与者id集合
		Set<String> participantIds = withJedis(jedis -> jedis.smembers(participantSetId));
		
		//	清空数据
		withJedis(jedis -> {
			Pipeline tx = jedis.pipelined();
			//	清空环境
			tx.del(contextRedisId);
			//	清空参与者集合
			tx.del(participantSetId);
			//	清空参与者
			if (participantIds != null && !participantIds.isEmpty()) {
				participantIds.forEach(pid -> tx.del(KeyNamespace.transactionParticipantId(biz, pid)));
			}
			tx.sync();
			return null;
		});
	}
	
	
	
	/****************************************************************************************************
	 * 
	 * 事务参与者相关操作
	 * 
	 ****************************************************************************************************/
	/**
	 * 保存参与者
	 * @throws ContextException 
	 */
	public void saveParticipant(IDistributedTransaction firstTx) {
		withJedis(jedis -> {
			Pipeline tx = jedis.pipelined();
			saveParticipant(tx, firstTx);
			tx.sync();
			return null;
		});
	}
	/**
	 * 保存参与者
	 */
	public void saveParticipant(Pipeline tx, IDistributedTransaction firstTx) {
		//	保存参与者对象
		String redisId = KeyNamespace.transactionParticipantId(firstTx.biz(), firstTx.participantId());
		tx.hset(redisId, KeyNamespace.FIELD_PARTICIPANT_ID, firstTx.participantId());
		tx.hset(redisId, KeyNamespace.FIELD_PARTICIPANT_TXID, firstTx.id());
		tx.hset(redisId, KeyNamespace.FIELD_PARTICIPANT_ROLE, firstTx.role().getCode());
		tx.hset(redisId, KeyNamespace.FIELD_PARTICIPANT_BIZ, firstTx.biz());
		tx.hset(redisId, KeyNamespace.FIELD_PARTICIPANT_STATUS, Integer.toString(firstTx.status().getVal()));
		long now = firstTx.startTimestamp();
		tx.hset(redisId, KeyNamespace.FIELD_PARTICIPANT_START_TIMESTAMP, Long.toString(now));
		tx.hset(redisId, KeyNamespace.FIELD_CONTEXT_START_DATE, dateFormat.format(new Date(now)));
		tx.expire(redisId, KeyNamespace.TIMEOUT);
		
		//	初始化参与者集合
		String setId = KeyNamespace.transactionParticipantSetIds(firstTx.biz(), firstTx.id());
		tx.sadd(setId, firstTx.participantId());
		tx.expire(setId, KeyNamespace.TIMEOUT);
	}
	/**
	 * 改写参与者当前状态
	 */
	public void updatePriticipantStatus(String biz, String txId, String pid, DistributedTransactionStatus status) {
		withJedis(jedis -> jedis.hset(KeyNamespace.transactionParticipantId(biz, pid), KeyNamespace.FIELD_PARTICIPANT_STATUS, Integer.toString(status.getVal())));
	}
	/**
	 * 改写参与者当前状态
	 */
	public void updatePriticipantStatus(Pipeline tx, String biz, String txId, String pid, DistributedTransactionStatus status) {
		tx.hset(KeyNamespace.transactionParticipantId(biz, pid), KeyNamespace.FIELD_PARTICIPANT_STATUS, Integer.toString(status.getVal()));
	}
	/**
	 * 获取参与者
	 * @param pids	参与者额原始id（不是redisid）
	 * @return	与pids下标对应的对象列表。中间可能存在null，当pid对应的redis取不到数据时为null
	 */
	@SuppressWarnings("unchecked")
	public List<TransactionParticipant> getParticipant(String biz, String... pids) {
		//	转为redisId
		List<String> redisIds = Arrays.stream(pids).map(pid -> KeyNamespace.transactionParticipantId(biz, pid)).collect(Collectors.toList());
		//	批量取
		List<Object> list = withJedis(jedis -> {
			Pipeline pipeline = jedis.pipelined();
			redisIds.forEach(rid -> pipeline.hgetAll(rid));
			return pipeline.syncAndReturnAll();
		});
		//	批量转
		if (list == null || list.isEmpty()) {return null;}
		List<TransactionParticipant> res = list.stream().map(o -> {
			if (o == null) {return null;}
			Map<String, String> map = (Map<String, String>) o;
			if (map == null || map.isEmpty()) {return null;} 
			
			TransactionParticipant t = new TransactionParticipant();
			t.setTxId(map.get(KeyNamespace.FIELD_PARTICIPANT_TXID));
			t.setParticipantId(KeyNamespace.FIELD_PARTICIPANT_ID);
			t.setBiz(map.get(KeyNamespace.FIELD_PARTICIPANT_BIZ));
			t.setRole(DistributedTransactionRole.getInstance(map.get(KeyNamespace.FIELD_PARTICIPANT_ROLE)));
			t.setStatus(DistributedTransactionStatus.getInstance(Integer.parseInt(map.get(KeyNamespace.FIELD_PARTICIPANT_STATUS))));
			t.setStartTimestamp(Long.parseLong(map.get(KeyNamespace.FIELD_PARTICIPANT_START_TIMESTAMP)));
			return t;
		}).filter(o -> o != null).collect(Collectors.toList());
		return res;
	}
	/**
	 * 清除参与者
	 */
	public void clearParticipant(Pipeline tx, String biz, String txId, String pid) {
		//	清除参与者集合
		tx.srem(KeyNamespace.transactionParticipantSetIds(biz, txId), pid);
		//	清除参与者
		tx.del(KeyNamespace.transactionParticipantId(biz, pid));
	}
	/**
	 * 清除参与者
	 */
	public void clearParticipant(String biz, String txId, String pid) {
		withJedis(jedis -> {
			Pipeline tx = jedis.pipelined();
			clearParticipant(tx, biz, txId, pid);
			tx.sync();
			return null;
		});
	}
	

	
	
	public JedisPool getPool() {
		return pool;
	}
	public void setPool(JedisPool pool) {
		this.pool = pool;
	}
	public DistributedTransactionConfigure.RedisConfigure getConf() {
		return conf;
	}
	public void setConf(DistributedTransactionConfigure.RedisConfigure conf) {
		this.conf = conf;
	}
}
